package com.shujia.useraction

import com.shujia.utils.{Constants, SparkMain, SparkTool}
import org.apache.spark.internal.Logging
import org.apache.spark.sql.functions.{concat, md5, upper}
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

class UserActionIPTV(day: String) extends UserActionBase {


  import spark.implicits._



  override def load(path: String, day: String): DataFrame = {
    spark
      .read
      .format("json")
      .load(s"$path/$day")

  }

  override def process(): Unit = {
    df = load(Constants.USER_ACTION_BASE_IPTV_PATH, day)
      .withColumn("uid", upper(md5(concat($"stbid", $"cid"))))
      .select($"uid", $"itemId", $"itemName", $"processDate", $"utctime")

  }

  override def save(): Unit = {
    df.write
      .format(Constants.USER_ACTION_IPTV_FORMAT)
      .mode(SaveMode.Overwrite)
      .save(s"${Constants.USER_ACTION_IPTV_PATH}/$day")

  }

  override def load(): DataFrame = {
    spark
      .read
      .format(Constants.USER_ACTION_IPTV_FORMAT)
      .load(s"${Constants.USER_ACTION_IPTV_PATH}/$day")

  }

}

object UserActionIPTV extends SparkMain {
  override def run(day: String): Unit = {
    val userActionIPTV: UserActionIPTV = UserActionIPTV(day)

    userActionIPTV.process()
    userActionIPTV.save()
  }

  def apply(day: String): UserActionIPTV = new UserActionIPTV(day)

}

